package com.heima.behavior.service.impl;

import com.alibaba.fastjson.JSON;
import com.heima.behavior.service.ReadService;
import com.heima.common.constants.BehaviorConstants;
import com.heima.common.constants.BehaviorMessageConstants;
import com.heima.common.constants.HotArticleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.behavior.dtos.ReadBehaviorDto;
import com.heima.model.common.dtos.ResponseResult;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.mess.UpdateArticleMess;
import lombok.RequiredArgsConstructor;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

@Service
@RequiredArgsConstructor
public class ReadServiceImpl implements ReadService {


    private final CacheService cacheService;

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final ThreadPoolTaskExecutor threadPoolTaskExecutor;

    private final String READ_BEHAVIOR_HASH_KEY = BehaviorConstants.ARTICLE_READ_COUNT;

    /**
     * 文章的浏览次数-使用hash结构存储
     *
     * @param dto
     * @return
     */
    @Override
    public ResponseResult readBehavior(ReadBehaviorDto dto) {
        // TODO:未做持久化
        // 1、检查参数
        if (dto == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
        // 2、新增到Redis
        saveViewCountToRedis(String.valueOf(dto.getArticleId()), dto.getCount());
        asyncKafkaStreamUpdateViewCountToDb(dto);
        return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
    }

    /**
     * 使用kafka stream 异步发送消息 进行数据聚合 更新数据库文章的浏览次数
     * @param dto 行为参数
     * @Author huangjianfei
     * @Since 2024/11/15
     * @Version day11
     */
    private void asyncKafkaStreamUpdateViewCountToDb(ReadBehaviorDto dto) {
        UpdateArticleMess mess=new UpdateArticleMess();
        mess.setArticleId(dto.getArticleId());
        mess.setType(UpdateArticleMess.UpdateArticleType.VIEWS);
        mess.setAdd(1);
        kafkaTemplate.send(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC,JSON.toJSONString(mess));
    }

    /**
     * 保存文章的浏览次数到redis
     * @param hashKey
     * @param count
     * @Author huangjianfei
     * @Since 2024/11/08
     * @Version day09
     */
    private void saveViewCountToRedis(String hashKey, Integer count) {
        // 1、查询redis是否存在
        if (cacheService.hGet(READ_BEHAVIOR_HASH_KEY, hashKey) == null) {
            cacheService.hPut(READ_BEHAVIOR_HASH_KEY, hashKey, String.valueOf(count));
        } else { // 不为null
            Long value = cacheService.hIncrBy(READ_BEHAVIOR_HASH_KEY, hashKey, count);
//             异步线程去做数据库的更新
//            if(value%5==0){
//                // 异步线程同步到Db中
//                   threadPoolTaskExecutor.submit(() ->{
//                       saveViewCountToDb(hashKey,value);
//                   });
//            }
        }
    }

    /**
     * 使用kafka异步发送消息去更新文章浏览数量
     * @param articleId 文章id
     * @param value 新增浏览量
     * @Author huangjianfei
     * @Since 2024/11/08
     * @Version day09
     */
    private void saveViewCountToDb(String articleId, Long value) {
        Map<String, Object> map = new HashMap<>();
        map.put("articleId", articleId);
        map.put("views", value);
        kafkaTemplate.send(BehaviorMessageConstants.APP_ARTICLE_READ_COUNT_TOPIC, JSON.toJSONString(map));
    }
}
